// @Author EthanScriptOn
// @Desc
package rocketmq_operator

import (
	"context"
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/common/identity"
	"gitee.com/fatzeng/srf_switch_expansion_package/connect"
	"gitee.com/fatzeng/srf_switch_expansion_package/extend_config"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/actuator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator"
	"gitee.com/fatzeng/srf_switch_stream_bridge/framework/operator/op_example"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

type RocketmqConsumeOperator struct {
	*op_example.LoopDetectionMultiChannelDataOperator
	*identity.IdentityInformationCarrier
}

func (r *RocketmqConsumeOperator) ExecuteWithLoop(ctx *actuator.SrfSwitchBridgeActuatorContext, operatorConfigIdentifier interface{}, structureParam interface{}) (err error) {
	operatorIdentifier, err := r.IdentityInformationCarrier.GenerateUniqueIdentifier()
	if err != nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error generating unique identifier: %v", err))
		return err
	}
	pipeline := r.GetDataPipeline(operatorIdentifier)
	if pipeline == nil {
		_ = ctx.Record(actuator.OperatorExecute, fmt.Sprintf("Error getting data pipeline: %v", err))
		return err
	}

	config := operator.GetOperatorConfig(operatorConfigIdentifier)
	if config == nil {
		return fmt.Errorf("config is nil")
	}
	operatorConfig, ok := config.(*extend_config.RocketmqConsumeOperatorConfig)
	if !ok {
		return fmt.Errorf("config is not *RocketmqConsumeOperatorConfig")
	}

	rocketmqConsumer, err := connect.GetRocketmqConsumer()
	err = rocketmqConsumer.Subscribe(operatorConfig.Topic, *operatorConfig.Selector, func(ctx context.Context, msg ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, extMsgTemp := range msg {
			extMsg := extMsgTemp
			pipeline <- extMsg
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		return fmt.Errorf("Subscribe error: %v\n", err)
	}
	select {}
}
